package com.raft.kvstore.client;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.net.Socket;

import com.alibaba.com.caucho.hessian.io.HessianInput;
import com.alibaba.com.caucho.hessian.io.HessianOutput;
import com.raft.core.node.NodeId;
import com.raft.core.service.AddNodeCommand;
import com.raft.core.service.Channel;
import com.raft.core.service.ChannelException;
import com.raft.core.service.RedirectException;
import com.raft.core.service.RemoveNodeCommand;
import com.raft.kvstore.MessageConstants;
import com.raft.kvstore.message.Failure;
import com.raft.kvstore.message.GetCommand;
import com.raft.kvstore.message.GetCommandResponse;
import com.raft.kvstore.message.Redirect;
import com.raft.kvstore.message.SetCommand;

public class SocketChannel implements Channel {

    private final String host;
    private final int port;

    public SocketChannel(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public Object send(Object payload) {
        try (Socket socket = new Socket()) {
            socket.setTcpNoDelay(true);
            socket.connect(new InetSocketAddress(this.host, this.port));
            this.write(socket.getOutputStream(), payload);
            return this.read(socket.getInputStream());
        } catch (IOException e) {
            throw new ChannelException("failed to send and receive", e);
        }
    }

    private Object read(InputStream input) throws IOException {
        DataInputStream dataInput = new DataInputStream(input);
        int messageType = dataInput.readInt();
        int payloadLength = dataInput.readInt();
        byte[] payload = new byte[payloadLength];
        dataInput.readFully(payload);
        switch (messageType) {
            case MessageConstants.MSG_TYPE_SUCCESS:
                return null;
            case MessageConstants.MSG_TYPE_FAILURE:
            	Failure protoFailure = decode(payload, Failure.class);
                throw new ChannelException("error code " + protoFailure.getErrorCode() + ", message " + protoFailure.getMessage());
            case MessageConstants.MSG_TYPE_REDIRECT:
            	Redirect protoRedirect = decode(payload, Redirect.class);
                throw new RedirectException(new NodeId(protoRedirect.getLeaderId()));
            case MessageConstants.MSG_TYPE_GET_COMMAND_RESPONSE:
            	GetCommandResponse protoGetCommandResponse = decode(payload, GetCommandResponse.class);
                if (!protoGetCommandResponse.isFound()) return null;
                return protoGetCommandResponse.getValue();
            default:
                throw new ChannelException("unexpected message type " + messageType);
        }
    }

    private void write(OutputStream output, Object payload) throws IOException {
        if (payload instanceof GetCommand) {
            this.write(output, MessageConstants.MSG_TYPE_GET_COMMAND, payload);
        } else if (payload instanceof SetCommand) {
            this.write(output, MessageConstants.MSG_TYPE_SET_COMMAND, payload);
        } else if (payload instanceof AddNodeCommand) {
            this.write(output, MessageConstants.MSG_TYPE_ADD_SERVER_COMMAND, payload);
        } else if (payload instanceof RemoveNodeCommand) {
            this.write(output, MessageConstants.MSG_TYPE_REMOVE_SERVER_COMMAND, payload);
        }
    }

    private void write(OutputStream output, int messageType, Object message) throws IOException {
    	ByteArrayOutputStream byteOutput = new ByteArrayOutputStream();
        HessianOutput hessianOutput = new HessianOutput(byteOutput);
        hessianOutput.writeObject(message);
        byte[] messageBytes = byteOutput.toByteArray();

        DataOutputStream dataOutput = new DataOutputStream(output);
        // 4(messageType) + 4(bytes.length) + VAR(bytes)
        dataOutput.writeInt(messageType);
        dataOutput.writeInt(messageBytes.length);
        dataOutput.write(messageBytes);
        dataOutput.flush();
    }

    private <T extends Serializable> T decode(byte[] bytes, Class<T> clazz) {
        if (bytes == null) {
            return null;
        }
        // 1、将字节数组转换成字节输入流
        ByteArrayInputStream bis = new ByteArrayInputStream(bytes);

        // step 1. 定义外部序列化工厂
        //ExtSerializerFactory extSerializerFactory = new ExtSerializerFactory();
        //extSerializerFactory.addSerializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisSerializer());
        //extSerializerFactory.addDeserializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisDeserializer());
        // step 2. 序列化工厂
        //SerializerFactory serializerFactory = new SerializerFactory();
        //serializerFactory.addFactory(extSerializerFactory);
        HessianInput hessianInput = new HessianInput(bis);
        //hessianInput.setSerializerFactory(serializerFactory);
        Object object = null;

        try {
            object = hessianInput.readObject();
        } catch (IOException e) {
        	e.printStackTrace();
        } catch (Exception e) {
        	e.printStackTrace();
        }

        return (T) object;
    }

}
